package com.deepglint.processfunction;

import com.deepglint.beans.SensorReading;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author mj
 * @version 1.0
 * @date 2021-11-23 21:57
 */
public class ProcessTest_ApplicationCase {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.getConfig().setAutoWatermarkInterval(200);

        DataStream<String> streamSource = env.socketTextStream("192.168.150.128", 7777);

        DataStream<SensorReading> dataStream = streamSource.map(line -> {
            String[] split = line.split(",");
            return new SensorReading(split[0], split[1], new Long(split[2]), new Double(split[3]));
        });

        SingleOutputStreamOperator<String> mapStream = dataStream.
                keyBy("id").
                process(new TempConsIncrWarning(10));

        mapStream.print();

        env.execute();
    }

    // 自定义处理逻辑，检测一段时间内温度持续上升，就报警
    static class TempConsIncrWarning extends KeyedProcessFunction<Tuple, SensorReading, String> {
        private Integer interval;
        private ValueState<Double> lastTempState;
        private ValueState<Long> timerTsState;

        public TempConsIncrWarning(Integer interval) {
            this.interval = interval;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>("my_temp", Double.class, 0.0));
            timerTsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("my_timer", Long.class));
        }

        @Override
        public void processElement(SensorReading value, KeyedProcessFunction<Tuple, SensorReading, String>.Context ctx, Collector<String> out) throws Exception {
            // 取出状态
            Double lastTemp = lastTempState.value();
            Long timerTs = timerTsState.value();

            // 如果温度上升且没有定时器的话
            if (value.getTemperature() > lastTemp && timerTs == null) {
                long ts = (ctx.timerService().currentProcessingTime()) + interval * 1000L;
                ctx.timerService().registerProcessingTimeTimer(ts);
                timerTsState.update(ts);
            } else if (value.getTemperature() < lastTemp && timerTs != null) {
                ctx.timerService().deleteProcessingTimeTimer(timerTs);
                timerTsState.clear();
            }

            //更新温度状态
            lastTempState.update(value.getTemperature());
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<Tuple, SensorReading, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
            System.out.printf("传感器%s温度持续%ss上升\n", ctx.getCurrentKey().getField(0), interval);
            timerTsState.clear();
        }

        @Override
        public void close() throws Exception {
            lastTempState.clear();
        }
    }
}
